from kafka import KafkaProducer, KafkaConsumer
import json
import time

def send_test_message():
    """发送测试消息"""
    try:
        producer = KafkaProducer(
            bootstrap_servers='172.16.12.182:6667',
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )

        test_message = {
            "message": "Test message for bfd",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "type": "test"
        }

        future = producer.send('ambari-kafka-service-check', test_message)
        result = future.get(timeout=10)
        print(f"消息发送成功! Partition: {result.partition}, Offset: {result.offset}")

        producer.flush()
        producer.close()
        return True

    except Exception as e:
        print("发送消息失败:", str(e))
        return False

def read_messages():
    """读取消息"""
    try:
        consumer = KafkaConsumer(
            'ambari-kafka-service-check',
            bootstrap_servers='172.16.12.182:6667',
            auto_offset_reset='earliest',
            enable_auto_commit=False,
            group_id=None,
            consumer_timeout_ms=5000
        )

        print("开始读取消息...")
        count = 0
        for message in consumer:
            try:
                msg_data = json.loads(message.value.decode('utf-8'))
                print(f"收到消息: {msg_data}")
            except:
                print(f"收到消息: {message.value.decode('utf-8')}")

            print(f"分区: {message.partition}, 偏移量: {message.offset}")
            count += 1
            if count >= 5:
                break

        if count == 0:
            print("没有读取到消息")

        consumer.close()
        return count

    except Exception as e:
        print("读取消息失败:", str(e))
        return 0

# 执行测试
if __name__ == "__main__":
    print("1. 发送测试消息...")
    if send_test_message():
        print("2. 等待2秒后读取消息...")
        time.sleep(2)
        read_messages()
from kafka import KafkaConsumer

try:
    # 先检查topic基本信息
    list_consumer = KafkaConsumer(bootstrap_servers='172.16.12.182:6667')
    topics = list_consumer.topics()
    print("可用topics:", topics)

    if 'ambari-kafka-service-check' in topics:
        partitions = list_consumer.partitions_for_topic('ambari-kafka-service-check')
        print(f"Topic分区数: {partitions}")

        # 检查是否有数据
        if partitions:
            from kafka.structs import TopicPartition
            topic_partitions = [TopicPartition('ambari-kafka-service-check', p) for p in partitions]
            end_offsets = list_consumer.end_offsets(topic_partitions)
            print(f"各分区末尾偏移量: {end_offsets}")

            total_messages = sum(offset for offset in end_offsets.values())
            print(f"Topic总消息数估计: {total_messages}")

    list_consumer.close()

except Exception as e:
    print("检查topic状态出错:", str(e))
from kafka import KafkaConsumer
from kafka.structs import TopicPartition

try:
    # 直接消费现有的 ambari-kafka-service-check topic
    consumer = KafkaConsumer(
        'ambari-kafka-service-check',  # 使用现有topic
        bootstrap_servers='172.16.12.182:6667',
        auto_offset_reset='earliest',
        enable_auto_commit=False,
        group_id=None
    )

    print("开始读取 'ambari-kafka-service-check' topic的数据...")
    count = 0
    for message in consumer:
        print(f"收到消息: {message.value.decode('utf-8')}")
        print(f"Topic: {message.topic}, Partition: {message.partition}, Offset: {message.offset}")
        count += 1
        if count >= 10:
            break
    if count == 0:
        print("该topic中没有数据")

except Exception as e:
    print("读取数据出错:", str(e))
